import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.CompareFilter;
import org.apache.hadoop.hbase.filter.RegexStringComparator;
import org.apache.hadoop.hbase.filter.RowFilter;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

/**
 * @Auther: zhezhe
 * @Date: 2018/11/2910:40
 * @Description:
 */
public class Student {
    Connection conn = null;
    Table table = null;

    @Before
    public void getConn() throws IOException {
        //构建一个连接对象
        Configuration conf = HBaseConfiguration.create();  //会自动加载hbase-site.xml
        conf.set("hbase.zookeeper.quorum", "master11:2181,slave1:2181,slave2:2181");
        conn = ConnectionFactory.createConnection(conf);
    }

    /**
     * 创建学生表
     */
    @Test
    public void createStudentTable() throws IOException {
        //从连接中构造一个DDL操作器
        Admin admin = conn.getAdmin();
        //创建一个表定义描述对象
        HTableDescriptor hTableDescriptor = new HTableDescriptor(TableName.valueOf("t_student"));
        //创建列族定义描述对象
        HColumnDescriptor hColumnDescriptor_1 = new HColumnDescriptor("base_info");
        HColumnDescriptor hColumnDescriptor_2 = new HColumnDescriptor("other_info");
        //设置该列族中存储数据的最大版本数，默认是1
        hColumnDescriptor_1.setMaxVersions(3);

        //将列族定义信息对象放入表定义对象中
        hTableDescriptor.addFamily(hColumnDescriptor_1);
        hTableDescriptor.addFamily(hColumnDescriptor_2);

        //用DDL操作器对象：admin来建表
        admin.createTable(hTableDescriptor);

        //关闭连接
        admin.close();
        conn.close();

    }

    /**
     * 创建课程表
     * @throws IOException
     */
    @Test
    public void createCourseTable() throws IOException {
        Admin admin = conn.getAdmin();
        //创建一个表定义描述对象
        HTableDescriptor hTableDescriptor = new HTableDescriptor(TableName.valueOf("t_course"));
        //创建列族定义描述对象
        HColumnDescriptor hColumnDescriptor_1 = new HColumnDescriptor("base_info");
        HColumnDescriptor hColumnDescriptor_2 = new HColumnDescriptor("other_info");
        //设置该列族中存储数据的最大版本数，默认是1
        hColumnDescriptor_1.setMaxVersions(3);

        //将列族定义信息对象放入表定义对象中
        hTableDescriptor.addFamily(hColumnDescriptor_1);
        hTableDescriptor.addFamily(hColumnDescriptor_2);

        //用DDL操作器对象：admin来建表
        admin.createTable(hTableDescriptor);

        //关闭连接
        admin.close();
        conn.close();
    }


    /**
     * 创建学生课程表
     * @throws IOException
     */
    @Test
    public void createSc() throws IOException {
        Admin admin = conn.getAdmin();
        //创建一个表定义描述对象
        HTableDescriptor hTableDescriptor = new HTableDescriptor(TableName.valueOf("t_sc"));
        //创建列族定义描述对象
        HColumnDescriptor hColumnDescriptor_1 = new HColumnDescriptor("base_info");
        HColumnDescriptor hColumnDescriptor_2 = new HColumnDescriptor("other_info");
        //设置该列族中存储数据的最大版本数，默认是1
        hColumnDescriptor_1.setMaxVersions(3);

        //将列族定义信息对象放入表定义对象中
        hTableDescriptor.addFamily(hColumnDescriptor_1);
        hTableDescriptor.addFamily(hColumnDescriptor_2);

        //用DDL操作器对象：admin来建表
        admin.createTable(hTableDescriptor);

        //关闭连接
        admin.close();
        conn.close();
    }



    /**
     * 删除表
     */

    @Test
    public void testDropTable() throws IOException {
        Admin admin = conn.getAdmin();
        //停用表
        admin.disableTable(TableName.valueOf("t_sc"));
        //删除表
        admin.deleteTable(TableName.valueOf("t_sc"));

        admin.close();
        conn.close();
    }

    /**
     * 往学生表添加数据
     * @throws IOException
     */
    @Test
    public void addStudentData() throws IOException {
        table = conn.getTable(TableName.valueOf("t_student"));
        ArrayList<Put> arrayList = new ArrayList<Put>();
        for (int i = 1; i < 5; i++) {
            Put put = new Put(Bytes.toBytes("s1"+i));
            put.addColumn(Bytes.toBytes("base_info"), Bytes.toBytes("sno"), Bytes.toBytes(String.valueOf(i)));
            put.addColumn(Bytes.toBytes("base_info"), Bytes.toBytes("sname"), Bytes.toBytes("student" + i));
            put.addColumn(Bytes.toBytes("base_info"), Bytes.toBytes("age"), Bytes.toBytes("17"));
            arrayList.add(put);
        }
        table.put(arrayList);
        arrayList.clear();
        for (int i = 5; i < 8; i++) {
            Put put1 = new Put(Bytes.toBytes("s2"+i));
            put1.addColumn(Bytes.toBytes("base_info"), Bytes.toBytes("sno"), Bytes.toBytes(String.valueOf(i)));
            put1.addColumn(Bytes.toBytes("base_info"), Bytes.toBytes("sname"), Bytes.toBytes("student" + i));
            put1.addColumn(Bytes.toBytes("base_info"), Bytes.toBytes("age"), Bytes.toBytes("16"));
            arrayList.add(put1);
        }
        table.put(arrayList);

    }


    /**
     * 往课程表添加数据
     * @throws IOException
     */
    @Test
    public void addCourseData() throws IOException {
        table = conn.getTable(TableName.valueOf("t_course"));
        Put put = new Put(Bytes.toBytes("c1"));
        put.addColumn(Bytes.toBytes("base_info"), Bytes.toBytes("cname"), Bytes.toBytes("Chinese"));
        put.addColumn(Bytes.toBytes("base_info"), Bytes.toBytes("tname"), Bytes.toBytes("zhangsan"));
        table.put(put);
        Put put1 = new Put(Bytes.toBytes("c2"));
        put1.addColumn(Bytes.toBytes("base_info"), Bytes.toBytes("cname"), Bytes.toBytes("Math"));
        put1.addColumn(Bytes.toBytes("base_info"), Bytes.toBytes("tname"), Bytes.toBytes("lisi"));
        table.put(put1);
        Put put2 = new Put(Bytes.toBytes("c3"));
        put2.addColumn(Bytes.toBytes("base_info"), Bytes.toBytes("cname"), Bytes.toBytes("English"));
        put2.addColumn(Bytes.toBytes("base_info"), Bytes.toBytes("tname"), Bytes.toBytes("wangwu"));
        table.put(put2);
        System.out.println("添加成功");
    }

    /**
     * 给学生课程表添加数据
     * @throws IOException
     */
    @Test
    public void addScData() throws IOException {
        table = conn.getTable(TableName.valueOf("t_sc"));
        Put put = new Put(Bytes.toBytes("s11_c1"));
        put.addColumn(Bytes.toBytes("base_info"), Bytes.toBytes("cno"), Bytes.toBytes("c1"));
        put.addColumn(Bytes.toBytes("base_info"), Bytes.toBytes("sno"), Bytes.toBytes("s11"));
        table.put(put);

        Put put1 = new Put(Bytes.toBytes("s12_c2"));
        put1.addColumn(Bytes.toBytes("base_info"), Bytes.toBytes("cno"), Bytes.toBytes("c2"));
        put1.addColumn(Bytes.toBytes("base_info"), Bytes.toBytes("sno"), Bytes.toBytes("s12"));
        table.put(put1);

        Put put2 = new Put(Bytes.toBytes("s11_c2"));
        put2.addColumn(Bytes.toBytes("base_info"), Bytes.toBytes("cno"), Bytes.toBytes("c2"));
        put2.addColumn(Bytes.toBytes("base_info"), Bytes.toBytes("sno"), Bytes.toBytes("s11"));
        table.put(put2);

        Put put3 = new Put(Bytes.toBytes("s12_c1"));
        put3.addColumn(Bytes.toBytes("base_info"), Bytes.toBytes("cno"), Bytes.toBytes("c1"));
        put3.addColumn(Bytes.toBytes("base_info"), Bytes.toBytes("sno"), Bytes.toBytes("s12"));
        table.put(put3);

        Put put5 = new Put(Bytes.toBytes("s13_c2"));
        put5.addColumn(Bytes.toBytes("base_info"), Bytes.toBytes("cno"), Bytes.toBytes("c2"));
        put5.addColumn(Bytes.toBytes("base_info"), Bytes.toBytes("sno"), Bytes.toBytes("s13"));
        table.put(put5);

        Put put6 = new Put(Bytes.toBytes("s25_c2"));
        put6.addColumn(Bytes.toBytes("base_info"), Bytes.toBytes("cno"), Bytes.toBytes("c2"));
        put6.addColumn(Bytes.toBytes("base_info"), Bytes.toBytes("sno"), Bytes.toBytes("s25"));
        table.put(put6);

        Put put7 = new Put(Bytes.toBytes("s26_c1"));
        put7.addColumn(Bytes.toBytes("base_info"), Bytes.toBytes("cno"), Bytes.toBytes("c1"));
        put7.addColumn(Bytes.toBytes("base_info"), Bytes.toBytes("sno"), Bytes.toBytes("s26"));
        table.put(put7);
    }


    /**
     * 查询某个学生所选的所有课程列表   s12
     * @throws IOException
     */
    @Test
    public void scanData() throws IOException {
        table = conn.getTable(TableName.valueOf("t_sc"));
        List<String> courseIds =new ArrayList<String>();
        String stuId="s12";
        //创建全表扫描的scan
        Scan scan = new Scan();


        //匹配rowkey以s12_开头的
        RowFilter rf1 = new RowFilter(CompareFilter.CompareOp.EQUAL, new RegexStringComparator(stuId+"_"));
        //设置过滤器
        scan.setFilter(rf1);
        //结果集
        ResultScanner rs1 = table.getScanner(scan);
        Iterator<Result> it = rs1.iterator();
        while(it.hasNext()){
            Result result = it.next();
            byte[] rowKey = result.getRow();
            courseIds.add(new String(rowKey));
        }

        for(String id : courseIds){
            Table  cTable = conn.getTable(TableName.valueOf("t_course"));
            String courseId = id.split("_")[1];
            Get get = new Get(courseId.getBytes());
            Result result= cTable.get(get);
            byte[] name = result.getValue("base_info".getBytes(), "cname".getBytes());
            System.out.println("课程ID:"+courseId+" 名称："+new String(name));
        }
    }

    /**
     * 按行键查询
     * @throws IOException
     */
    @Test
    public void quaryRowkey() throws IOException {
        table = conn.getTable(TableName.valueOf("t_sc"));
        Get get = new Get(Bytes.toBytes("s11_c1"));
        Result re = table.get(get);
        System.out.println(Bytes.toString(re.getValue(Bytes.toBytes("base_info"), Bytes.toBytes("cno"))));
        System.out.println(Bytes.toString(re.getValue(Bytes.toBytes("base_info"),Bytes.toBytes("sno"))));
    }


    /**
     * 查询选修某个课程c2的学生列表
     * @throws IOException
     */
    @Test
    public  void findStudentByCourse() throws IOException {
        table = conn.getTable(TableName.valueOf("t_sc"));
        String courseId = "c2";
        List<String> studentIds = new ArrayList<String>();
        Scan scan = new Scan();
        RowFilter rf1 = new RowFilter(CompareFilter.CompareOp.EQUAL,new RegexStringComparator("_"+courseId));
        scan.setFilter(rf1);
        ResultScanner rs1 = table.getScanner(scan);
        Iterator<Result> it = rs1.iterator();
        while (it.hasNext()){
            Result result = it.next();
            byte[] rowKey = result.getRow();
            studentIds.add(new String(rowKey));
        }

        for(String id : studentIds){
            String stuId = id.split("_")[0];
            Table  stuTable = conn.getTable(TableName.valueOf("t_student"));
            Get get = new Get(stuId.getBytes());
            Result result= stuTable.get(get);
            byte[] name = result.getValue("base_info".getBytes(), "sname".getBytes());
            System.out.println("课程ID:"+courseId+" 学生名字："+new String(name));
        }
    }



    @Test
    public void updateCourse() throws IOException {
        String stuId="s12";
        String courseId = "c2";
        String scId=stuId+"_"+courseId;
        Delete delete = new Delete(scId.getBytes());
        Table table = conn.getTable(TableName.valueOf("t_sc"));
        table.delete(delete);
    }

}
